package com.thlws.study.redismq.internal;

import com.thlws.study.redismq.internal.helper.RedisMqHelper;
import com.thlws.study.redismq.internal.queue.Queue;
import com.thlws.study.redismq.internal.queue.QueueSelector;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author HanleyTang 2019-08-07
 */
@Component
public class MessageReceiver {

    private final static Logger logger = LoggerFactory.getLogger(MessageReceiver.class);

    @Autowired
    private RedisMqHelper mqHelper;

    @Resource(name = "myRedisTemplate")
    private RedisTemplate redisTemplate;

    private ThreadPoolExecutor threadPoolExecutor;

    private final int timeout = 2;

    @PostConstruct
    public void init(){

        BasicThreadFactory basicThreadFactory = new BasicThreadFactory.Builder().namingPattern("message-receiver-pool-%d").daemon(true).build();
        threadPoolExecutor = new ThreadPoolExecutor(3, 3,
                0L, TimeUnit.MILLISECONDS,
                new SynchronousQueue<Runnable>(),basicThreadFactory);

        threadPoolExecutor.submit(this::queueOne);
        threadPoolExecutor.submit(this::queueTwo);
        threadPoolExecutor.submit(this::queueThree);

//        threadPoolExecutor.submit(this::channelOne);
//        threadPoolExecutor.submit(this::channelTwo);
//        threadPoolExecutor.submit(this::channelThree);

    }

    private void queueOne() {
        Queue queue = QueueSelector.getInstance().select(0);
        while (true) {


            try {
                Object object = redisTemplate.opsForList().rightPopAndLeftPush(queue.getQueueName(),queue.getBakQueueName(),timeout,TimeUnit.SECONDS);
                if (ObjectUtils.isEmpty(object)) {
                    continue;
                }
                logger.info("thread:{}, 队列1 拿到消息:{}",Thread.currentThread().getName(),object);
                //TODO 处理具体业务

                //处理成功后备份队列中移除
                redisTemplate.opsForList().remove(queue.getBakQueueName(), 0, object);
                logger.info(" queue1 处理完成");
            } catch (Exception e) {
                logger.error("处理消息失败");
            }
        }
    }


    private void queueTwo() {
        Queue queue = QueueSelector.getInstance().select(1);
        while (true) {

            Object object = redisTemplate.opsForList().rightPopAndLeftPush(queue.getQueueName(),queue.getBakQueueName(),timeout,TimeUnit.SECONDS);
            if (ObjectUtils.isEmpty(object)) {
                continue;
            }
            logger.info("thread:{}, 队列2 拿到消息:{}",Thread.currentThread().getName(),object);
            //TODO 处理具体业务

            //处理成功后备份队列中移除
            redisTemplate.opsForList().remove(queue.getBakQueueName(), 0, object);
            logger.info(" queue2 处理完成");
        }
    }


    private void queueThree() {
        Queue queue = QueueSelector.getInstance().select(2);
        while (true) {

            Object object = redisTemplate.opsForList().rightPopAndLeftPush(queue.getQueueName(),queue.getBakQueueName(),timeout,TimeUnit.SECONDS);
            if (ObjectUtils.isEmpty(object)) {
                continue;
            }
            logger.info("thread:{}, 队列3 拿到消息:{}",Thread.currentThread().getName(),object);
            //TODO 处理具体业务

            //处理成功后备份队列中移除
            redisTemplate.opsForList().remove(queue.getBakQueueName(), 0, object);
            logger.info(" queue3 处理完成");
        }
    }




    private void channelOne() {
        mqHelper.subscribe("cctv-1", (message -> {
            String name = Thread.currentThread().getName();
            logger.info("thread={},收到 cctv-1 消息: {}",name, message);
        }));
    }

    private void channelTwo() {
        mqHelper.subscribe("cctv-2", (message -> {
            String name = Thread.currentThread().getName();
            logger.info("thread={},收到 cctv-2 消息: {}", name,message);
        }));
    }

    private void channelThree() {
        mqHelper.subscribe("cctv-3", (message -> {
            String name = Thread.currentThread().getName();
            logger.info("thread={},收到 cctv-3 消息: {}", name,message);
        }));
    }


    @PreDestroy
    public void destroy(){
        if (null != threadPoolExecutor && !threadPoolExecutor.isShutdown()){
            threadPoolExecutor.shutdown();
        }
    }
}
